package com.jydw.da.frame.listen;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import com.jydw.da.frame.mapper.UserMapper;
import com.jydw.da.frame.pojo.CanalMessage;
import com.jydw.da.frame.pojo.User;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Canal + RabbitMQ 监听数据库数据变化
 *
 * @author chenyj
 * @date 2023/7/4 23:14
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class CanalListener {
    @Autowired
    private UserMapper userMapper;

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "canal.queue", durable = "true"),
                    exchange = @Exchange(value = "canal.exchange"),
                    key = "canal.routing.key"
            )
    })
    public void handleDataChange(CanalMessage<User> canalMessage, Message message, Channel channel) {
        log.info("canalMessage: {}", canalMessage);
        log.info("message: {}", message);
        log.info("Channel: {}", channel);
        List<User> data = canalMessage.getData();
        if("Insert".equalsIgnoreCase(canalMessage.getType()) && "user_copy1".equalsIgnoreCase(canalMessage.getTable())){
            data.forEach(x -> {
                userMapper.insert(x);
            });
        }
//        CanalMessage canalMessage = JSONObject.parseObject(message, CanalMessage.class);
//        String tableName = canalMessage.getTable();
//        log.info("Canal 监听 {} 发生变化；明细：{}", tableName, message);
    }
}
